package cn.bawei.shop.realtime.etl.`trait`
import cn.bawei.shop.realtime.etl.utils.KafkaProps
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._

abstract class MQBaseETL(env:StreamExecutionEnvironment) extends BaseETL[String]{
  /**
   * 从kafka中读取数据，传递返回的数据类型
   *
   * @param topic
   * @return
   */
  override def getKafkaDataSream(topic: String): DataStream[String] = {
    //创建消费者对象，从kafka中消费数据，消费到的数据是字符串类型
    val kafkaProducer: FlinkKafkaConsumer011[String] = new FlinkKafkaConsumer011[String](
      topic,
      new SimpleStringSchema(),
      KafkaProps.getKafkaProperties()
    )

    //将消费者对象添加到数据源
    val logDataStream: DataStream[String] = env.addSource(kafkaProducer)

    //返回消费到的数据
    logDataStream
  }

}
